package com.example.kafka.controller;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.TopicListing;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@RestController
public class KafkaProducer {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Resource
    private KafkaAdmin kafkaAdmin;

    /**
     * 发送消息
     *
     * @param normalMessage
     */
    @GetMapping("/kafka/normal/{topicName}/{message}")
    public void sendMessage1(@PathVariable("topicName") String topicName, @PathVariable("message") String normalMessage) {
        kafkaTemplate.send(topicName, normalMessage);
    }

    /**
     * 获取topic
     */
    @GetMapping("/kafka/topicName")
    public String topicName() {
        //1.获取Kafka配置信息
        Map<String, Object> configs = kafkaAdmin.getConfigurationProperties();
        //2.创建客户端AdminClient
        AdminClient adminClient = KafkaAdminClient.create(configs);
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        String collect = null;
        try {
            collect = listTopicsResult.listings().get().stream().map(result -> result.name()).collect(Collectors.joining(","));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return collect;
    }


    /**
     * 获取topic
     */
    @GetMapping("/deletTopic/{topicName}")
    public String deletTopic(@PathVariable("topicName") String topicName) {
        //1.获取Kafka配置信息
        Map<String, Object> configs = kafkaAdmin.getConfigurationProperties();
        //2.创建客户端AdminClient
        AdminClient adminClient = KafkaAdminClient.create(configs);
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        String returnString = null;
        try {
            Collection<TopicListing> topicListings = listTopicsResult.listings().get();

            String collect = topicListings.stream().map(result -> result.name()).collect(Collectors.joining(","));
            if (listTopicsResult.listings().get().stream().anyMatch(result -> result.name().equals(topicName))) {

                adminClient.deleteTopics(Arrays.asList(topicName));
                returnString = "删除成功!";
                listTopicsResult = adminClient.listTopics();
                topicListings = listTopicsResult.listings().get();
                collect = topicListings.stream().map(result -> result.name()).collect(Collectors.joining(","));
            } else {
                returnString = "当前topic不存在!";
            }
            returnString += "【" + collect + "】";
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return returnString;
    }
}